package com.atguigu.flink.chapter10.query;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/6/29 09:09
 */
public class QueryWindowTVFDemo1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        
        // 在 ddl 中定义时间属性
        tEnv.executeSql("create table sensor(" +
                            " id string, " +
                            " ts bigint, " +
                            " vc int, " +
                            " et as to_timestamp_ltz(ts, 3)," +
                            " watermark for et as et - interval '1' second " +
                            ")with(" +
                            "  'connector' = 'filesystem', " +
                            "  'path' = 'input/sensor.txt', " +
                            "  'format' = 'csv' " +
                            ")");
    
    
        /*tEnv.sqlQuery("select " +
                          "id, window_start, window_end, " +
                          "sum(vc) vc_sum " +
                          "from table( tumble( table sensor, descriptor(et), interval '5' second ) )" +
                          "group by id, window_start, window_end")
            .execute()
            .print();*/
    
    
       /* tEnv.sqlQuery("select " +
                          "id, window_start, window_end, " +
                          "sum(vc) vc_sum " +
                          "from table( hop( table sensor, descriptor(et),interval '2' second, interval '6' second ) )" +
                          "group by id, window_start, window_end")
            .execute()
            .print();*/
        
        // 累积窗口:
        // 每隔 1h 计算下当天的 pv
        
        //
        tEnv.sqlQuery("select " +
                          "id, window_start, window_end, " +
                          "sum(vc) vc_sum " +
                          "from table( cumulate( table sensor, descriptor(et),interval '5' second, interval '20' second ) )" +
                          "group by rollup(id, window_start, window_end)")
            .execute()
            .print();
        
       
        
    }
}